#ifdef _MSC_VER
#pragma warning(disable:4244) // for google protobuf
#endif

#include "PBCodec.h"
#include <google/protobuf/descriptor.h>
#include "TcpSocketOps.h"
#include "Log.h"

namespace dsth {

PBCodec::PBCodec(MessageHandler msg_handler, ErrorMsgHandler error_handler) 
    : msg_handler_(msg_handler),
    error_handler_(error_handler) {
}

void PBCodec::OnMessage(Connection* conn, ChannelBuffer* buf, Timestamp receiveTime) {
     while (buf->ReadableBytes() >= kHeaderLen) {
         const int32_t len = buf->PeekInt32();
         if (len > kMaxMessageLen) {
             error_handler_(conn, buf, receiveTime);
             break;
         }
         if (buf->ReadableBytes() >= static_cast<size_t>(len)) {
             ErrorCode errorcode = kNoError;
             Message* msg = Parse(buf->Peek() + kHeaderLen, len, &errorcode);
             if (errorcode == kNoError && msg) {
                 msg_handler_(conn, msg, receiveTime);
                 buf->Retrieve(kHeaderLen + len);
                 delete msg;
             }
             else {
                 error_handler_(conn, buf, receiveTime);
                 buf->RetrieveAll();
             }
         }
         else {
             break;
         }
     }
}

void PBCodec::SendMessage(const Connection* conn, const Message* message) {
    ChannelBuffer buff;
    const std::string& typeName = message->GetTypeName();
    int32_t nameLen = static_cast<int32_t>(typeName.size());
    buff.AppendInt32(nameLen);
    buff.Append(typeName.c_str(), nameLen);

    GOOGLE_DCHECK(message->IsInitialized()) << "google serialize error";
    int byteSize = message->ByteSize();
    buff.EnsureCapacity(byteSize);
    uint8_t* start = reinterpret_cast<uint8_t*>(buff.GetWriter());
    uint8_t* end = message->SerializeWithCachedSizesToArray(start);
    if (end - start != byteSize) {
        LOG_FATAL("google::protobuf::Message SerializeWithCachedSizesToArray failed during serialization.");
    }
    buff.HasWritten(byteSize);
    int32_t len = net::H2N_32(static_cast<int32_t>(buff.ReadableBytes()));
    buff.Perpend(&len, sizeof len);
    const_cast<Connection*>(conn)->Send(buff);
}

Message* PBCodec::Parse(const char* data, int32_t len, ErrorCode* errorcode) {
    int32_t nameLen = 0;
    memcpy(&nameLen, data , sizeof nameLen);
    nameLen = net::N2H_32(nameLen);
    Message* msg = NULL;
    if (nameLen && nameLen < len) {
        std::string msg_name(data + sizeof nameLen, data + sizeof nameLen + nameLen);
        msg = CreateMessage(msg_name);
        if (msg) {
            // parse from buffer
            const char* parseData = data + sizeof nameLen + nameLen;
            int32_t dataLen = len - nameLen - sizeof nameLen;
            if (msg->ParseFromArray(parseData, dataLen)) {
                *errorcode = kNoError;
            }
            else {
                *errorcode = kParseError;
            }
        }
        else {
            *errorcode = kUnknownMessageType;
        }
    }
    else {
        *errorcode = kInvalidNameLen;
    }
    return msg;
}

Message* PBCodec::CreateMessage(const std::string& name) {
    google::protobuf::Message* message = NULL;
    const google::protobuf::Descriptor* descriptor =
        google::protobuf::DescriptorPool::generated_pool()->FindMessageTypeByName(name);
    if (descriptor) {
        const google::protobuf::Message* prototype =
            google::protobuf::MessageFactory::generated_factory()->GetPrototype(descriptor);
        if (prototype) {
            message = prototype->New();
        }
    }
    return message;
}

} // namespace
